package com.ccb.channel.p1;

import java.io.IOException;
import java.net.URLDecoder;
import java.sql.Timestamp;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.log4j.Logger;

import com.alibaba.fastjson.JSON;
import com.ccb.filter.Filters;
import com.ccb.filter.RecordFilter;
import com.ccb.util.DateUtil;
import com.ccb.util.StandardRecord;
import com.ccb.util.XmlJSON;

public class P1Mapper extends Mapper<Object, Text, Text, Text> {
	/**
	 * Logger for this class
	 */
	private static final Logger logger = Logger.getLogger(P1Mapper.class);

	private MultipleOutputs<Text, Text> mos = null;
	private RecordFilter recordFilter ; 
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		if (logger.isDebugEnabled()) {
			logger.debug("setup(Context) - start");
		}
		recordFilter = Filters.getFilter("p1");
		if(recordFilter == null){
			logger.info("fiter is null");
		}
		mos = new MultipleOutputs<Text, Text>(context);

		if (logger.isDebugEnabled()) {
			logger.debug("setup(Context) - end");
		}
	}

	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		if (logger.isDebugEnabled()) {
			logger.debug("cleanup(Context) - start");
		}

		mos.close();

		if (logger.isDebugEnabled()) {
			logger.debug("cleanup(Context) - end");
		}
	}

	public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
		if (logger.isDebugEnabled()) {
			logger.debug("map(Object, Text, Context) - start");
		}

		String content = value.toString().trim();

		try {
			if (content.isEmpty() || !StringUtils.contains(content, "senddata="))return;
			StandardRecord record = new StandardRecord();
			String sendData = StringUtils.substringBetween(content, "senddata=[", "]!");
			sendData=URLDecoder.decode(sendData, "utf-8");
			String dateStr= StringUtils.substringBetween(sendData, "<transaction_data>", "</transaction_data>");
			String timeStr=StringUtils.substringBetween(sendData, "<transaction_time>", "</transaction_time>");
			String dateTimeStr=dateStr+timeStr;
			String formatStr=DateUtil.date2Str("yyyy-MM-dd HH:mm:ss.SSS", DateUtil.str2Date("yyyyMMddHHmmss",dateTimeStr));
			Timestamp time=Timestamp.valueOf(formatStr);
			String serviceId = StringUtils.substringBetween(content, "ptxcode:", " ");
			if(recordFilter != null && !recordFilter.accept(serviceId)){
				if (logger.isDebugEnabled()) {
					logger.debug("map(Object, Text, Context) - end");
				}
				return;
			}
			String backData = StringUtils.substringBetween(content, "backdata=[", "]!");
			record.setServiceId(serviceId);
			record.setTime(time);
			record.setRequest(sendData);
			record.setResponse(backData);
			record.setResponseData(XmlJSON.xmlToJSONObj(backData));
			record.setRequestData(XmlJSON.xmlToJSONObj(sendData));
			mos.write( new Text(), new Text(JSON.toJSONString(record)), record.getServiceId());
			// context.write(new Text(), new Text(JSON.toJSONString(record)));
		} catch (Exception e) {
			String path=((FileSplit)context.getInputSplit()).getPath().toString();
			logger.error("record is error:"+path+""+key.toString()+"errorType="+e.getClass()+"errorSimpleInfo"+e.getLocalizedMessage());
			logger.error("map(Object, Text, Context)", e);
			if (logger.isDebugEnabled()) {
				logger.debug("map(Object, Text, Context) - end");
			}
			return;
		}

		if (logger.isDebugEnabled()) {
			logger.debug("map(Object, Text, Context) - end");
		}
	}
}
